package org.example.lookup_join;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkNormalLookupJoin {
    public static void main(String[] args) throws Exception {
        // 配置Flink Web UI端口
        Configuration config = new Configuration();
        config.setString("rest.bind-port", "8081"); // 设置Web UI端口为8081

        // 设置流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 创建Paimon目录
        String createCatalogSQL = "CREATE CATALOG paimon WITH (" +
                "    'type' = 'paimon'," +
                "    'warehouse' = 'file:///tmp/paimon'" +
                ");";
        System.out.println("Executing SQL: " + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);

        // 创建订单表（使用datagen生成数据）
        String createOrderTableSQL = "CREATE TABLE IF NOT EXISTS order_table (" +
                "    order_id BIGINT, " + // 订单ID
                "    user_id BIGINT, " + // 用户ID，用于关联用户表
                "    order_amount DOUBLE, " + // 订单金额
                "    order_date TIMESTAMP(3), " + // 订单日期
                "    proc_time AS PROCTIME() " + // 使用处理时间
                ") WITH (" +
                "    'connector' = 'datagen', " + // 使用datagen连接器生成数据
                "    'rows-per-second' = '10', " + // 每秒生成的行数
                "    'fields.user_id.min' = '1', " + // user_id的最小值，与user_table中的user_id范围一致
                "    'fields.user_id.max' = '100', " + // user_id的最大值，与user_table中的user_id范围一致
                "    'fields.order_amount.min' = '10', " + // 订单金额的最小值
                "    'fields.order_amount.max' = '1000'" + // 订单金额的最大值
                ");";
        System.out.println("Executing SQL: " + createOrderTableSQL);
        tableEnv.executeSql(createOrderTableSQL);

        // 执行lookup join查询
        String lookupJoinSQL = "SELECT " +
                "    o.order_id, " + // 订单ID
                "    o.user_id, " + // 用户ID
                "    u.user_name, " + // 用户名
                "    u.user_email, " + // 用户邮箱
                "    o.order_amount, " + // 订单金额
                "    o.order_date " + // 订单日期
                "FROM order_table AS o " + // 订单表
                "JOIN `paimon`.`default`.`user_table` FOR SYSTEM_TIME AS OF o.proc_time AS u " + // 与user_table进行lookup join
                "ON o.user_id = u.user_id;"; // 连接条件
        System.out.println("Executing SQL: " + lookupJoinSQL);
        tableEnv.executeSql(lookupJoinSQL).print();

        // 启动流任务
        env.execute("Flink Normal Lookup Join");
    }
}